package preprocess;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.UUID;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;

public class PvSearchClickJoiner extends Configured implements Tool {
	private static final int maxPvSearchSpan = 15;
	private static final int maxClickSearchSpan = 1800;
	public static String getPvSearchClickDetailPath(String date) {
		//return "D:\\dev\\clickprediction\\data\\PvSearchClickDetailLog";
		return Utils.rootDir + date + "/PvSearchClickDetailLog/";
	}
	
	enum MapCounter{PV_INVALID, CLICK_INVALID, SEARCH_INVALID, PV_TOTAL, CLICK_TOTAL, SEARCH_TOTAL};
	enum JoinCounter{PV_0_CLICK_0_SEARCH_0, PV_0_CLICK_N_SEARCH_0, PV_0_CLICK_0_SEARCH_N, PV_0_CLICK_N_SEARCH_N, 
					PV_N_CLICK_0_SEARCH_0, PV_N_CLICK_N_SEARCH_N, PV_N_CLICK_N_SEARCH_0, PV_N_CLICK_0_SEARCH_N,
					JOIN_3, JOIN_3_UNMATCHED_SEARCH, JOIN_3_UNMATCHED_PV, JOIN_3_MATCHED_EXTACTLY, JOIN_3_MATCHED_MULTIPLE, 
					JOIN_2, JOIN_2_UNMATCHED, JOIN_2_MATCHED_EXTACTLY, JOIN_2_MATCHED_MULTIPLE,
					EXCEED_MAX_COUNT,
					JOIN_COUNT, PV_RECORDS, CLICK_RECORDS, CLICK_RECORDS_TO_JOIN, CLICK_RECORDS_JOINED, SEARCH_RECORDS, REDUCE_INVALID_INPUT,
					};
					
	public static class PvMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		
		private static ArrayList<Integer> joinColumns = null;
		private static ArrayList<Integer> getJoinColumns() {
			
			if (joinColumns == null) {
				joinColumns = new ArrayList<Integer>();
				//userid, query, sort
				joinColumns.add(3);
				joinColumns.add(4);
				joinColumns.add(7);				
			}
			return joinColumns;
		}		
		
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] pv = line.split(Utils.separator);
			reporter.incrCounter(MapCounter.PV_TOTAL, 1);
			
			if (pv.length == 9) {
				ArrayList<Integer> joinKeys = getJoinColumns();
				String[] keys = new String[joinKeys.size()];				
				
				for (int i = 0; i < joinKeys.size(); ++i) {
					keys[i] = pv[joinKeys.get(i)];				
				}
				k.set(Utils.getRecord(keys, Utils.separator));
				
				output.collect(k, value);
			} else {
				reporter.incrCounter(MapCounter.PV_INVALID, 1);
			}
			
		}
	}

	public static class ClickMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();

		private static ArrayList<Integer> joinColumns = null;		
		
		private static ArrayList<Integer> getJoinColumns() {
			
			if (joinColumns == null) {
				joinColumns = new ArrayList<Integer>();				
				joinColumns.add(0);
				joinColumns.add(1);
				joinColumns.add(11);
			}
			return joinColumns;
		}
		
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] click = line.split(Utils.separator);
			reporter.incrCounter(MapCounter.CLICK_TOTAL, 1);
			if (click.length != 27) {
				reporter.incrCounter(MapCounter.CLICK_INVALID, 1);
				return;
			}
				
			ArrayList<Integer> joinKeys = getJoinColumns();
			
			String[] keys = new String[joinKeys.size()];
			
			for (int i = 0; i < joinKeys.size(); ++i) {
				keys[i] = click[joinKeys.get(i)];					
			}
			
			k.set(Utils.getRecord(keys, Utils.separator));			
			output.collect(k, value);
		}
			
	}
	
	public static class SearchMap extends MapReduceBase implements
	Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();

		private static ArrayList<Integer> joinColumns = null;		

		private static ArrayList<Integer> getJoinColumns() {
			
			if (joinColumns == null) {
				joinColumns = new ArrayList<Integer>();				
				joinColumns.add(0);
				joinColumns.add(1);
				joinColumns.add(11);
			}
			return joinColumns;
		}

		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] click = line.split(Utils.separator);
			reporter.incrCounter(MapCounter.SEARCH_TOTAL, 1);
			if (click.length != 25) {
				reporter.incrCounter(MapCounter.SEARCH_INVALID, 1);
				return;
			}
				
			ArrayList<Integer> joinKeys = getJoinColumns();
			
			String[] keys = new String[joinKeys.size()];
			
			for (int i = 0; i < joinKeys.size(); ++i) {
				keys[i] = click[joinKeys.get(i)];					
			}
			
			k.set(Utils.getRecord(keys, Utils.separator));			
			output.collect(k, value);
		}
	
	}

	
	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {
		private static class TimeComparator implements Comparator<String[]> {
			public TimeComparator(int column) {
				this.column = column;
			}
			private int column;
			public int compare(String[] str1, String[] str2) {			
				Date date1 = Utils.getDate(str1[column]);
				Date date2 = Utils.getDate(str2[column]);
				return (int)(date1.getTime() - date2.getTime());
				
			}
		}
		private boolean isClickSearchMatched(String[] click, String[] search) {
			
			//bucket, 用户分类，machineid, 搜索类目,上一步点击模块, 列表模式，消保筛选，所在地筛选,第几页
			int[] indexes = new int[]{2, 4, 5, 9, 10, 12, 13, 14, 16};
			//todo: suozaidigeshi
			for (int i = 0; i < indexes.length; ++i) {
				if (!click[indexes[i]].equalsIgnoreCase(search[indexes[i]]))
					return false;
			}
			//Tab类型
			if (!click[25].equalsIgnoreCase(search[24]))
				return false;
			//点击时间和search时间
			if (!Utils.isLaterThan(click[7], search[7]))
				return false;
			return true;
		}
		
		private boolean isClickPvSearchMatched(String[] click, String[] pv,  String[] search) {
			int pos = Integer.parseInt(click[17]);
			//itemid
			String itemId = click[18];
			//attribute
			String attribute = search[17];
			if (!attribute.equalsIgnoreCase(pv[6]))
				return false;
			//点击时间，pv时间
			if (!Utils.isLaterThan(click[7], pv[2]))
				return false;
			//search时间，pv时间
			if (Utils.getTimeSpan(search[7], pv[2]) > maxPvSearchSpan)
				return false;
			//点击时间，search时间
			if (Utils.getTimeSpan(click[7], search[7]) > maxClickSearchSpan)
				return false;
			//String category = click[19];			
			//if (!(category.equalsIgnoreCase("0") && pv[5].isEmpty() || category.equals(pv[5])))
			//	return false;
			//是否为豆腐块
			int isDoufu = Integer.parseInt(click[26]);
			String[] itemLists = pv[8].split(Utils.pvAuctionSeparator1);
			String[] doufuItems;
			String[] items;
			if (itemLists.length == 2) {
				doufuItems = itemLists[0].split(Utils.pvAuctionSeparator2);
				items = itemLists[1].split(Utils.pvAuctionSeparator2);
			}
			else {
				doufuItems = new String[]{};
				items = itemLists[0].split(Utils.pvAuctionSeparator2);
			}
			if (isDoufu == 1 && doufuItems.length >= pos && doufuItems[pos - 1].equalsIgnoreCase(itemId)
					||
					isDoufu == 0 && pos > doufuItems.length && items.length >= pos - doufuItems.length && items[pos - doufuItems.length - 1].equalsIgnoreCase(itemId))
				return true;
			else return false;			
		}
		
		private boolean isPvSearchMatched(String[] pv,  String[] search) {
			String attribute = search[17];
			if (!attribute.equalsIgnoreCase(pv[6]))
				return false;
			if (Utils.getTimeSpan(search[7], pv[2]) > maxPvSearchSpan)
				return false;
			return true;	
		}
		
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			ArrayList<String[]> pvContainer = new ArrayList<String[]>();
			ArrayList<String[]> clickContainer = new ArrayList<String[]>();
			ArrayList<String[]> searchContainer = new ArrayList<String[]>();
			int count = 0;
			while (values.hasNext()) {
				if (++count > 300) {
					reporter.incrCounter(JoinCounter.EXCEED_MAX_COUNT, 1);
					return;
				}
				String result = values.next().toString();
				String[] strs = result.split(Utils.separator);
				if (strs.length == 9) {					
					pvContainer.add(strs);
				} else if (strs.length == 27) {	
					//所在地转换
					strs[14] = strs[14].replace("\\x", "%");
					strs[14] = strs[14].replace(",", "%2C");
					strs[14] = strs[14].toUpperCase();
					clickContainer.add(strs);
				} else if (strs.length == 25) {
					searchContainer.add(strs);
				} else {
					reporter.incrCounter(JoinCounter.REDUCE_INVALID_INPUT, 1);
				}
				
			}
			java.util.Collections.sort(pvContainer, new TimeComparator(2));
			java.util.Collections.sort(clickContainer, new TimeComparator(7));
			java.util.Collections.sort(searchContainer, new TimeComparator(7));
			
			reporter.incrCounter(JoinCounter.JOIN_COUNT, 1);
			reporter.incrCounter(JoinCounter.PV_RECORDS, pvContainer.size());
			reporter.incrCounter(JoinCounter.CLICK_RECORDS, clickContainer.size());
			reporter.incrCounter(JoinCounter.SEARCH_RECORDS, searchContainer.size());
			HashSet<Integer> pvOccupied = new HashSet<Integer>();
			HashSet<Integer> clickOccupied = new HashSet<Integer>();
			HashSet<Integer> searchOccupied = new HashSet<Integer>();
			if (pvContainer.isEmpty()) {
				if (clickContainer.isEmpty() && searchContainer.isEmpty())
					reporter.incrCounter(JoinCounter.PV_0_CLICK_0_SEARCH_0, 1);
				else if (clickContainer.isEmpty() && !searchContainer.isEmpty())
					reporter.incrCounter(JoinCounter.PV_0_CLICK_0_SEARCH_N, 1);
				else if (!clickContainer.isEmpty() && searchContainer.isEmpty())
					reporter.incrCounter(JoinCounter.PV_0_CLICK_N_SEARCH_0, 1);
				else reporter.incrCounter(JoinCounter.PV_0_CLICK_N_SEARCH_N, 1);
				return;
			} else if (clickContainer.isEmpty() && searchContainer.isEmpty()) {
				reporter.incrCounter(JoinCounter.PV_N_CLICK_0_SEARCH_0, 1);
				return;
			} else if (!clickContainer.isEmpty() && searchContainer.isEmpty()){
				reporter.incrCounter(JoinCounter.PV_N_CLICK_N_SEARCH_0, 1);
				return;
				
			} else if (clickContainer.isEmpty() && !searchContainer.isEmpty()) {
				reporter.incrCounter(JoinCounter.PV_N_CLICK_0_SEARCH_N, 1);
				
			} else {
				reporter.incrCounter(JoinCounter.CLICK_RECORDS_TO_JOIN, clickContainer.size());
				reporter.incrCounter(JoinCounter.PV_N_CLICK_N_SEARCH_N, 1);
				reporter.incrCounter(JoinCounter.JOIN_3, pvContainer.size());
				//collect starting from pv
				for (int i = 0; i < pvContainer.size(); ++i) {
					String[] pv = pvContainer.get(i);
					ArrayList<Integer> matchedSearch = new ArrayList<Integer>();
					ArrayList<ArrayList<Integer>> matchedClickWithSearch = new ArrayList<ArrayList<Integer>>();
					int matchedCount = 0;
					for (int j = 0; j < searchContainer.size(); ++j) {
						if (searchOccupied.contains(j))
							continue;
						String[] search = searchContainer.get(j);
						if (isPvSearchMatched(pv, search)) {
							ArrayList<Integer> matchedClick = new ArrayList<Integer>();							
							for (int k = 0; k < clickContainer.size(); ++k) {
								if (clickOccupied.contains(k))
									continue;
								String[] click = clickContainer.get(k);
								
								if (isClickSearchMatched(click, search) && isClickPvSearchMatched(click, pv, search)) {
									matchedClick.add(k);
								}
							}
							if (matchedClick.size() > 0) {								
								matchedSearch.add(j);
								matchedClickWithSearch.add(matchedClick);
								++matchedCount;
							}
						}
					}
					
					if (matchedCount == 1) {
						reporter.incrCounter(JoinCounter.JOIN_3_MATCHED_EXTACTLY, 1);
						//output.collect(new Text(""), new Text(Utils.getRecord(pv, Utils.separator)));
					} else if (matchedCount > 1) {
						reporter.incrCounter(JoinCounter.JOIN_3_MATCHED_MULTIPLE, 1);
					}
					else {
						reporter.incrCounter(JoinCounter.JOIN_3_UNMATCHED_PV, 1);
						//output.collect(new Text(""), new Text(Utils.getRecord(click, Utils.separator)));
					}
					String[] search = null;
					int j = 0; 
					for (;j < matchedSearch.size(); ++j) {
						String[] tmp = searchContainer.get(matchedSearch.get(j));
						if (Utils.isLaterThan(tmp[7], pv[2])) {
							search = tmp;
							break;
						}
					}
					if (matchedSearch.size() != 0 && search == null) {
						j = matchedSearch.size() - 1;						
						search = searchContainer.get(matchedSearch.get(j));					
					}
					
					if (search != null) {
						pvOccupied.add(i);
						searchOccupied.add(matchedSearch.get(j));
						ArrayList<Integer> matchedClick = matchedClickWithSearch.get(j);
						for (int k = 0; k < matchedClick.size(); ++k) {
							clickOccupied.add(matchedClick.get(k));
						}
						reporter.incrCounter(JoinCounter.CLICK_RECORDS_JOINED, matchedClick.size());
						joinClickPvSearch(clickContainer, matchedClick, pv, search, output);
					}
					
				}
			}
			
			//collect pv search join

			for (int i = 0; i < pvContainer.size(); ++i) {
				if (pvOccupied.contains(i))
					continue;
				reporter.incrCounter(JoinCounter.JOIN_2, 1);
				String[] pv = pvContainer.get(i);
				ArrayList<Integer> matchedSearch = new ArrayList<Integer>();
				for (int j = 0; j < searchContainer.size(); ++j) {
					if (searchOccupied.contains(j))
						continue;
					String[] search = searchContainer.get(j);
					if (isPvSearchMatched(pv, search))
						matchedSearch.add(j);
				}
				if (matchedSearch.size() == 0) {
					reporter.incrCounter(JoinCounter.JOIN_2_UNMATCHED, 1);
					//output.collect(new Text(""), new Text(Utils.getRecord(pv, Utils.separator)));
				} else if (matchedSearch.size() == 1) {
					reporter.incrCounter(JoinCounter.JOIN_2_MATCHED_EXTACTLY, 1);
				} else {
					reporter.incrCounter(JoinCounter.JOIN_2_MATCHED_MULTIPLE, 1);
				}
				String[] search = null;
				int j = 0; 
				for (;j < matchedSearch.size(); ++j) {
					String[] tmp = searchContainer.get(matchedSearch.get(j));
					if (Utils.isLaterThan(tmp[7], pv[2])) {
						search = tmp;
						break;
					}
				}
				if (matchedSearch.size() != 0 && search == null) {
					j = matchedSearch.size() - 1;					
					search = searchContainer.get(matchedSearch.get(j));					
				}
				if (search != null) {
					searchOccupied.add(matchedSearch.get(j));
					joinPvSearch(pv, search, output);
				}
				
			}
		}
		
		private String getPvStr(String[] pv) {
			//pv时间，直达后类目，商品idlist
			String pvStr = Utils.getRecord(new String[]{pv[2], pv[5], pv[8]}, Utils.separator);
			return pvStr;
		}
		
		private String getClickStr(String[] click) {
			//点击时间，搜索的来源pid, 点击位置，点击商品id(数字)，是否为豆腐块商品
			String clickStr = Utils.getRecord(
					new String[]{click[7], click[8], click[17], click[18], click[26]}, 
					"\u0001");
			return clickStr;
		}
		
		/*
		 * 
		 * 1.search[1~25]
		 * 26.UUID
		 * 27.pv时间
		 * 28.直达后类目
		 * 29.商品idlist
		 * 30.clicklist[点击时间^A搜索的来源pid^A点击位置^A点击商品id(数字)^A是否为豆腐块商品^B....]
		 */
		private void joinClickPvSearch(ArrayList<String[]> clicks, ArrayList<Integer> matchedClick, String[] pv, String[] search, OutputCollector<Text, Text> output)  throws IOException{
			String searchStr = Utils.getRecord(search, Utils.separator);
			String pvStr = getPvStr(pv);
			StringBuilder clickBuilder = new StringBuilder();
			if (matchedClick.size() > 0)
				clickBuilder.append(getClickStr(clicks.get(matchedClick.get(0))));
			for (int i = 1; i < matchedClick.size(); ++i) {
				clickBuilder.append("\u0002");
				String[] click = clicks.get(matchedClick.get(i));	
				clickBuilder.append(getClickStr(click));
			}
			String key = searchStr;
			String value = Utils.getRecord(new String[]{UUID.randomUUID().toString(), pvStr, clickBuilder.toString()}, Utils.separator);
			output.collect(new Text(key), new Text(value));		
		}
		
		private void joinPvSearch(String[] pv, String[] search, OutputCollector<Text, Text> output) throws IOException{
			String searchStr = Utils.getRecord(search, Utils.separator);
			String pvStr = getPvStr(pv);
			String clickStr = "-";
			String key = searchStr;
			String value = Utils.getRecord(new String[]{UUID.randomUUID().toString(), pvStr, clickStr}, Utils.separator);
			output.collect(new Text(key), new Text(value));		
		}
		
	}

	/*
	 * (non-Javadoc)
	 * @see org.apache.hadoop.util.Tool#run(java.lang.String[])
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/PvSearchClickDetailLog/
	 * uuid
	 * search
	 * pv
	 * clicks
	 */
	public int run(String[] args) throws Exception {
	
		// input
		String pvLogFile = DumpPvLog.getPvLogPath(args[0]);
		String clickLogFile = ClickLogItemIdConvertor.getClickLogPath(args[0]);
		String searchLogFile = DumpSearchLog.getSearchLogPath(args[0]);
		
		// Output
		String output = getPvSearchClickDetailPath(args[0]);
		System.out.println("start job PvSearchClickJoiner...");
		System.out.println("pvLogFile: " + pvLogFile);
		System.out.println("clickLogFile: " + clickLogFile);
		System.out.println("searchLogFile: " + searchLogFile);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JobConf conf = new JobConf(getConf(), PvSearchClickJoiner.class);
		conf.setJobName("PvSearchClickJoiner");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, 
				new Path(pvLogFile), 
				TextInputFormat.class, 
				PvMap.class);
		
		MultipleInputs.addInputPath(conf, 
				new Path(clickLogFile),
				TextInputFormat.class,
				ClickMap.class);
		
		MultipleInputs.addInputPath(conf, 
				new Path(searchLogFile),
				TextInputFormat.class,
				SearchMap.class);
		FileOutputFormat.setOutputPath(conf, new Path(output));
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		int res = ToolRunner.run(new Configuration(), new PvSearchClickJoiner(),
				args);
		System.exit(res);
	}

}